package com.hn.naixue;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.FileOutputStream;

/**
 * @Author: Qingwr
 * @Date: 2022/8/11 21:24
 * @Description hdfs API案例
 */
public class Demo {

    /**
     * 创建目录
     */
    public static void mkdirs(String s, String path) throws Exception {
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        fileSystem.mkdirs(new Path(path));
        fileSystem.close();
    }

    /**
     * 删除文件/目录
     */
    public static void rmdir(String path) throws Exception {
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        fileSystem.delete(new Path(path), true);
    }

    /**
     * 遍历文件
     */
    public static void listFiles() throws Exception {
        //1:获取FileSystem
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        RemoteIterator<LocatedFileStatus> iterator = fileSystem.listFiles(new Path("/"), true);
        //3、遍历迭代器
        while (iterator.hasNext()) {
            LocatedFileStatus fileStatus = iterator.next();
            //获取文件的绝对路径 : hdfs://hadoop0:8020/xxx
            System.out.println(fileStatus.getPath() + "======" + fileStatus.getPath().getName()); //文件的block信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) {
                    System.out.println("主机为: " + host);
                }
            }
            System.out.println("block数量为: " + blockLocations.length);
        }
    }


    /**
     * @param localPath : "C:/Users/DELL/Desktop/temp.txt"
     * @param hdfsPath  "/hdfsApi2"
     * @deprecated 文件上传
     */
    public static void uploadFile(String localPath, String hdfsPath) throws Exception {
        //1:获取FileSystem
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        //2:调用方法，实现上传
        fileSystem.copyFromLocalFile(new Path(localPath), new Path(hdfsPath));
        //3:关闭FileSystem
        fileSystem.close();
    }

    public static void downloadFile(String hdfsPath, String localPath) throws Exception {
        //1:获取FileSystem
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        fileSystem.copyToLocalFile(new Path(hdfsPath), new Path(localPath));
        //3:关闭FileSystem
        fileSystem.close();
    }

    public static void downloadFile2(String hdfsPath, String localPath) throws Exception {
        //1:获取FileSystem
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        // 2:获取hdfs的输入流
        FSDataInputStream inputStream = fileSystem.open(new Path(hdfsPath));
        //3:获取本地路径的输出流
        FileOutputStream outputStream = new FileOutputStream(localPath);
        //4:文件的拷贝
        IOUtils.copy(inputStream, outputStream);
        //5:关闭流
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);
        fileSystem.close();
    }

    public static void mergeUploadFile(String src, String dest) throws Exception {
        //1:获取FileSystem
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        //2:获取hdfs大文件的输出流
        FSDataOutputStream outputStream = fileSystem.create(new Path(dest));
        //3:获取一个本地文件系统
        LocalFileSystem localFileSystem = FileSystem.getLocal(new Configuration());
        //4:获取本地文件夹下所有文件的详情
        FileStatus[] fileStatuses = localFileSystem.listStatus(new Path(src));
        //5:遍历每个文件，获取每个文件的输入流
        for (FileStatus fileStatus : fileStatuses) {
            FSDataInputStream inputStream = localFileSystem.open(fileStatus.getPath());
            //6:将小文件的数据复制到大文件
            IOUtils.copy(inputStream, outputStream);
            IOUtils.closeQuietly(inputStream);
        }
        //7:关闭流
        IOUtils.closeQuietly(outputStream);
        localFileSystem.close();
        fileSystem.close();
    }

    public static boolean isFile(String path) throws Exception {
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        return fileSystem.isFile(new Path(path));
    }

    public static boolean isDir(String path) throws Exception {
        FileSystem fileSystem = FileSystemFactory.getFileSystem3();
        return fileSystem.isDirectory(new Path(path));
    }
}